package com.hgvip.stream;

import com.hgvip.common.KafkaParams;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStreamBuilder;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by arnold.zhu on 2017/8/3.
 */
public class KafkaStream {

    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream1");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaParams.BOOTSTRAP_SERVERS);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        StreamsConfig config = new StreamsConfig(props);

        KStreamBuilder builder = new KStreamBuilder();
        builder.stream("my-topic24").mapValues(value -> ((String) value).length() + "").to("my-topic25");

        KafkaStreams streams = new KafkaStreams(builder, config);
        streams.start();
    }

}
